package com.djhu.hbase;
/**
 * Created by zw on 2017/12/21.
 */

import org.apache.commons.collections.CollectionUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.data.hadoop.hbase.HbaseTemplate;
import org.springframework.data.hadoop.hbase.RowMapper;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * code is far away from bug with the animal protecting
 * ┏┓　　　┏┓
 * ┏┛┻━━━┛┻┓
 * ┃　　　　　　　┃
 * ┃　　　━　　　┃
 * ┃　┳┛　┗┳　┃
 * ┃　　　　　　　┃
 * ┃　　　┻　　　┃
 * ┃　　　　　　　┃
 * ┗━┓　　　┏━┛
 * 　　┃　　　┃神兽保佑
 * 　　┃　　　┃代码无BUG！
 * 　　┃　　　┗━━━┓
 * 　　┃　　　　　　　┣┓
 * 　　┃　　　　　　　┏┛
 * 　　┗┓┓┏━┳┓┏┛
 * 　　　┃┫┫　┃┫┫
 * 　　　┗┻┛　┗┻┛
 *
 * <p>
 * ---------------------------------
 */
public class HBaseOperator implements HbaseOperation{
    private Configuration configuration;
    private HbaseTemplate hbaseTemplate;

    private String famliyName = "BIGDATA";
    @Override
    public boolean createTable(String tableName, List<String> fields) {
//        hbaseTemplate.execute()
        Connection connection = null;
        Admin admin = null;

        try {
            connection = ConnectionFactory.createConnection(configuration);
            admin = connection.getAdmin();
           // admin.isTableAvailable()
            if (!admin.isTableAvailable(TableName.valueOf(tableName))) {
                HTableDescriptor hbaseTable = new HTableDescriptor(TableName.valueOf(tableName));
                hbaseTable.addFamily(new HColumnDescriptor(famliyName));
                admin.createTable(hbaseTable);
            }
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        } finally {
            try {
                if (admin != null) {
                    admin.close();
                }

                if (connection != null && !connection.isClosed()) {
                    connection.close();
                }
            } catch (Exception e2) {
                e2.printStackTrace();
            }
        }
    }

    @Override
    public String originalput(String tableName,String rowkey, Map<String, Object> map){
        HTable hTable = null;
        try {
            hTable = new HTable(configuration,TableName.valueOf(tableName));
            List<Put> putList = new ArrayList<>();
            for (Map.Entry<String, Object> entry:map.entrySet()){
                Put put = new Put(Bytes.toBytes(rowkey));
                put.addColumn(Bytes.toBytes(this.famliyName),Bytes.toBytes(entry.getKey()),Bytes.toBytes(String.valueOf(entry.getValue())));
                putList.add(put);
            }
            hTable.put(putList);
        } catch (IOException e) {
            e.printStackTrace();
        }finally {
            try {
                if (hTable != null){
                    hTable.close();
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        return rowkey;
    }

    public int bacthput(String tableName,List<HbaseEntity> hbaseEntitys){
        HTable hTable = null;
        int size = 0;
        try {
            hTable = new HTable(configuration, TableName.valueOf(tableName));
            List<Put> putList = new ArrayList<>();
            for (HbaseEntity hbaseEntity:hbaseEntitys){
                for (Map.Entry<String, Object> entry:hbaseEntity.getDataMap().entrySet()){
                    Put put = new Put(Bytes.toBytes(hbaseEntity.getRowKey()));
                    put.addColumn(Bytes.toBytes(this.famliyName),Bytes.toBytes(entry.getKey()),Bytes.toBytes(String.valueOf(entry.getValue())));
                    putList.add(put);
                    //1000条提交一次 防止数据太大 溢出
                    if (putList.size() > 1000){
                        hTable.put(putList);
                        putList.clear();
                    }
                }
                size++;
            }
            hTable.put(putList);
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            if (hTable != null){
                try {
                    hTable.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
        return size;
    }

    @Override
    public String put(String table,String rowkey, Map<String, Object> map){
        for (Map.Entry<String, Object> entry:map.entrySet()){
            hbaseTemplate.put(table,rowkey,this.famliyName,entry.getKey(), Bytes.toBytes(String.valueOf(entry.getValue())));
        }
        return rowkey;
    }


    @Override
    public Map<String, Object> get(String tableName, String rowKey) {
        Map<String,Object> data=hbaseTemplate.get(tableName, rowKey, new RowMapper<Map<String,Object> >() {
            @Override
            public Map<String,Object> mapRow(Result result, int i) throws Exception {
                Map<String,Object> map = new HashMap<>();
                List<Cell> cells = result.listCells();
                if (CollectionUtils.isEmpty(cells)){
                    return map;
                }
                for (Cell cell:cells){
                    String key = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength()).toLowerCase();
                    String dataValue = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                    map.put(key,dataValue);
                }
                return map;

            }

        });
        return data;
    }


    @Override
    public Map<String, Object> get(String tableName, String rowKey, List<String> fields) {
        List<String> caselessfileds = new ArrayList<>();
        for (String field:fields){
            caselessfileds.add(field);
        }
        Map<String,Object> map = new HashMap<>();
        for (Map.Entry<String, Object> entry:get(tableName, rowKey).entrySet()){
            if (fields.contains(entry.getKey())){
                map.put(entry.getKey(),entry.getValue());
            }
        }
        return map;
    }

    public <T> T get(String tableName,String rowKey,Class<T> tClass){
        PropertyRowMapper<T> propertyRowMapper = new PropertyRowMapper<T>(tClass);
        return (T)hbaseTemplate.get(tableName,rowKey,this.famliyName,propertyRowMapper);
    }

    @Override
    public Boolean delete(String tableName, String rowKey) {
        hbaseTemplate.delete(tableName,rowKey,famliyName);
        return true;
    }

    public Configuration getConfiguration() {
        return configuration;
    }

    public void setConfiguration(Configuration configuration) {
        this.configuration = configuration;
    }

    public HbaseTemplate getHbaseTemplate() {
        return hbaseTemplate;
    }

    public void setHbaseTemplate(HbaseTemplate hbaseTemplate) {
        this.hbaseTemplate = hbaseTemplate;
    }

    public String getFamliyName() {
        return famliyName;
    }

    public void setFamliyName(String famliyName) {
        this.famliyName = famliyName;
    }
}